package com.shujia.core

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel

object Demo22Cache {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("Demo22Cache")

    val sc: SparkContext = new SparkContext(conf)

    val stuRDD: RDD[String] = sc.textFile("spark/data/stu/students.txt")

    val newStuRDD: RDD[String] = stuRDD.map(line => {
      // 通过打印的次数判断数据有没有被Spark缓存起来
      println("进入了map方法")
      line
    })
    // 对多次使用的RDD可以进行缓存
    //    newStuRDD.cache()
    // cache相当于默认使用MEMORY_ONLY的缓存策略
    //    newStuRDD.persist(StorageLevel.MEMORY_ONLY)
    /**
     * 如何选择合适的缓存策略：
     * 内存充足：MEMORY_ONLY
     * 内存不够：MEMORY_AND_DISK_SER
     * 缓存策略需要通过persist方法进行指定
     */
    newStuRDD.persist(StorageLevel.MEMORY_AND_DISK_SER)

    // 统计班级人数
    newStuRDD.map(line => (line.split(",")(4), 1)).reduceByKey(_ + _).foreach(println)

    // 统计性别人数
    newStuRDD.map(line => (line.split(",")(3), 1)).reduceByKey(_ + _).foreach(println)

    // 统计年龄分布情况
    newStuRDD.map(line => (line.split(",")(2).toInt, 1)).reduceByKey(_ + _).foreach(println)

    while (true) {

    }

    // 用完需要释放缓存
    newStuRDD.unpersist()

  }

}
